package com.cryptoneedle.center.source.service.metadata;


import com.baomidou.dynamic.datasource.annotation.DS;
import com.bubbles.engine.data.core.repository.BaseRepository;
import com.bubbles.engine.data.core.service.BaseService;
import com.cryptoneedle.center.source.entity.metadata.SourceCatalog;
import com.cryptoneedle.center.source.entity.metadata.SourceDatabase;
import com.cryptoneedle.center.source.entity.metadata.SourceJdbc;
import com.cryptoneedle.center.source.entity.metadata.SourceTable;
import com.cryptoneedle.center.source.enums.SourceEnabledReason;
import com.cryptoneedle.center.source.key.metadata.SourceTableKey;
import com.cryptoneedle.center.source.properties.SourceProperties;
import com.cryptoneedle.center.source.properties.SshProperties;
import com.cryptoneedle.center.source.repository.metadata.SourceTableRepository;
import com.cryptoneedle.center.source.service.constraint.*;
import com.cryptoneedle.center.source.service.enhance.SourceEnhanceDimensionService;
import com.cryptoneedle.center.source.service.enhance.SourceEnhanceDorisAnalyzeColumnService;
import com.cryptoneedle.center.source.service.enhance.SourceEnhanceDorisAnalyzeTableService;
import com.cryptoneedle.center.source.service.enhance.SourceEnhanceIncrementService;
import com.cryptoneedle.center.source.util.JdbcUtil;
import com.cryptoneedle.center.source.util.SshManager;
import com.google.common.collect.Maps;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * <p>description: 源数据源-表 </p>
 *
 * @author CryptoNeedle
 * @date 2025-06-20
 */
@DS("pg")
@Slf4j
@Service
public class SourceTableService extends BaseService<SourceTable, SourceTableKey> {
    
    @Resource
    private SourceTableRepository sourceTableRepository;
    
    @Override
    public BaseRepository<SourceTable, SourceTableKey> getRepository() {
        return this.sourceTableRepository;
    }
    
    @Resource
    private SourceJdbcService sourceJdbcService;
    
    @Resource
    private SourceServerService sourceServerService;
    
    @Resource
    private SourceCatalogService sourceCatalogService;
    
    @Resource
    private SourceDatabaseService sourceDatabaseService;
    
    @Resource
    private SourceTableService sourceTableService;
    
    @Resource
    private SourceColumnService sourceColumnService;
    
    @Resource
    private SourceConstraintPrimaryService sourceConstraintPrimaryService;
    
    @Resource
    private SourceConstraintUniqueService sourceConstraintUniqueService;
    
    @Resource
    private SourceConstraintForeignService sourceConstraintForeignService;
    
    @Resource
    private SourceConstraintCheckService sourceConstraintCheckService;
    
    @Resource
    private SourceIndexUniqueService sourceIndexUniqueService;
    
    @Resource
    private SourceEnhanceDorisAnalyzeTableService sourceEnhanceDorisAnalyzeTableService;
    
    @Resource
    private SourceEnhanceDorisAnalyzeColumnService sourceEnhanceDorisAnalyzeColumnService;
    
    @Resource
    private SourceEnhanceIncrementService sourceEnhanceIncrementService;
    
    @Resource
    private SourceEnhanceDimensionService sourceEnhanceDimensionService;
    
    @Resource
    private SshProperties sshProperties;
    
    @Resource
    private SourceProperties sourceProperties;
    
    @Resource
    private SshManager sshManager;
    
    @Transactional
    public List<SourceTable> listByDatabase(SourceDatabase sourceDatabase) {
        return sourceTableRepository.listByDatabase(sourceDatabase.getSourceCatalogName(), sourceDatabase.getSourceDatabaseName());
    }
    
    @Transactional
    public List<SourceTable> listEnabled() {
        return sourceTableRepository.listEnabled();
    }
    
    @Transactional
    public void addList(List<SourceTable> sourceTables) {
        sourceTableService.saveAll(sourceTables);
    }
    
    @Transactional
    public void saveList(List<SourceTable> sourceTables) {
        sourceTableService.saveAll(sourceTables);
    }
    
    @Transactional
    public void deleteList(List<SourceTable> sourceTables) {
        sourceTableService.deleteAll(sourceTables);
    }
    
    @Transactional
    public void syncSource(SourceCatalog sourceCatalog, SourceDatabase sourceDatabase) {
        SourceJdbc sourceJdbc = sourceCatalog.getSourceJdbc();
        String sourceCatalogName = sourceCatalog.getSourceCatalogName();
        String sourceDatabaseName = sourceDatabase.getSourceDatabaseName();
        
        JdbcTemplate jdbcTemplate = JdbcUtil.getJdbcTemplate(sourceJdbc);
        List<SourceTable> sourceTables = jdbcTemplate.query(sourceJdbc
                                                                    .getDatabaseType()
                                                                    .getDatabase()
                                                                    .getTableSql(sourceDatabaseName), (resultSet, rowNum) -> new SourceTable()
                .setSourceCatalogName(sourceCatalogName)
                .setSourceSystemCode(sourceDatabase.getSourceSystemCode())
                .setSourceDatabaseName(sourceDatabaseName)
                .setSourceTableName(resultSet.getString("sourceTableName"))
                .setSourceTableComment(resultSet.getString("sourceTableComment"))
                .setSourceTableCommentManual(false)
                .setSourceTableEnabled(false)
                .setSourceTableEnabledReason(SourceEnabledReason.AUTO)
                .setSourceDimensionName(null)
                .setSourceStatistic(new SourceTable.SourceStatistic()
                                            .setSourceStatisticRowCount(resultSet.getLong("sourceStatisticRowCount"))
                                            .setSourceStatisticDatetime(resultSet.getDate("sourceStatisticDatetime")))
                .setDorisStatistic(null)
                .setDetailStatistic(null)
                .setTransform(null)
                .setDoris(null)
                .setSourceDatabase(sourceDatabase)
                .setSourceColumns(null)
                .setSourceConstraintPrimaries(null)
                .setSourceConstraintUniques(null)
                .setSourceConstraintForeigns(null)
                .setSourceConstraintChecks(null)
                .setSourceIndexUniques(null)
                .setSourceEnhanceIncrements(null)
                .setSourceEnhanceDimensions(null)
                .setSourceEnhanceDorisAnalyzeTables(null));
        
        sourceTableService.compareSource(sourceDatabase, sourceTables);
    }
    
    @Transactional
    public void compareSource(SourceDatabase sourceDatabase, List<SourceTable> dealList) {
        List<SourceTable> originList = sourceDatabase.getSourceTables();
        // 已存在
        Map<String, SourceTable> originMap = Maps.uniqueIndex(originList, SourceTable::getSourceTableName);
        // 待处理
        Map<String, SourceTable> dealMap = Maps.uniqueIndex(dealList, SourceTable::getSourceTableName);
        
        // 新增
        List<SourceTable> extraList = dealList.stream().filter(deal -> !originMap.containsKey(deal.getSourceTableName())).toList();
        
        // 保存
        List<SourceTable> existsList = originList.stream().filter(origin -> dealMap.containsKey(origin.getSourceTableName())).peek(origin -> {
            SourceTable deal = dealMap.get(origin.getSourceTableName());
            if (!origin.getSourceTableCommentManual()) {
                origin.setSourceTableComment(deal.getSourceTableComment());
            }
            origin.setSourceStatistic(deal.getSourceStatistic());
        }).toList();
        
        // 移除
        List<SourceTable> missList = originList.stream().filter(item -> !dealMap.containsKey(item.getSourceTableName())).toList();
        
        sourceTableService.addList(extraList);
        sourceTableService.saveList(existsList);
        sourceTableService.deleteList(missList);
        
        log.info("[SYNC] {} -> {} -> 元数据采集 -> Table              => 共计{} | 新增{} | 删除{}", sourceDatabase.getSourceCatalogName(), sourceDatabase.getSourceDatabaseName(), String.format("%5d", dealList.size()), String.format("%5d", extraList.size()), String.format("%5d", missList.size()) + (
                !missList.isEmpty() ? " => " + missList.stream().map(SourceTable::getSourceTableName).collect(Collectors.joining(",")) : ""));
    }
}